package com.ycy;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;

import java.util.HashMap;
import java.util.Map;


@Component
@Slf4j
public class ConsumerDemo {

//  @KafkaListener(
//          topics = {"fnw_logging_kafka_topic"},
//          groupId = "ConsumerDemo5",
//          errorHandler = "consumerAwareErrorHandler"
//  )
//  public void consumer(String value, Acknowledgment ack) {
//    log.info("value {}", value);
//  }

  @KafkaListener(
          topics = {"fnw_logging_kafka_topic"},
          groupId = "ConsumerDemo5",
          errorHandler = "consumerAwareErrorHandler"
  )
  public void consumer(String value) {
    log.info("value {}", value);
  }


  /**
   * 异常处理器
   */
  @Bean
  public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
    return (message, exception, consumer) -> {
      System.out.println("消费异常：" + message.getPayload());
      exception.printStackTrace();
      return null;
    };
  }


}
